package cn.com.pc.content.service;

import cn.com.pc.content.constant.TagConstants;
import cn.com.pc.content.domain.tag.Tag;
import cn.com.pc.content.domain.tag.TagRepository;
import cn.com.pc.content.domain.tag.TagTree;
import cn.com.pc.content.dto.TagDTO;
import cn.com.pc.content.service.es.ESCurdService;
import cn.com.pc.content.service.es.ElasticEntity;
import cn.com.pc.content.util.JsonUtils;
import com.fasterxml.jackson.core.type.TypeReference;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

@Slf4j
@Service
@RequiredArgsConstructor
public class RebuildEsTagIndexService {

    private final TagRepository tagRepository;

    private final TagTreeService tagTreeService;

    private final ESCurdService esCurdService;

    private final ThreadPoolTaskExecutor taskExecutor;

    // 每批次查询mysql写入es的数量
    @Value("${rebuildEsIndex.pageSize}")
    private int pageSize;

    // 每个线程任务处理的数量
    @Value("${rebuildEsIndex.eachCursorSize}")
    private int eachCursorSize;


    public void rebuild() {
        //mysql数据 导出到es
        try {
            StopWatch watch = new StopWatch();
            watch.start("重建标签索引" + TagConstants.TAG_IDX);
            //clearEsIndex();
            exportMysqlDocToEsIndex(taskExecutor);
            watch.stop();
            log.info( "{}任务已完成，执行时间{}ms", watch.getLastTaskName(), watch.getLastTaskTimeMillis());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void clearEsIndex() {
        QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();
        esCurdService.deleteByQuery(TagConstants.TAG_IDX, queryBuilder);
    }

    private void exportMysqlDocToEsIndex(ThreadPoolTaskExecutor executor) {
        List<Future<Integer>> futures = new ArrayList<>();
        long documentSize = tagRepository.count();
        long connectCursorTime = documentSize % eachCursorSize > 0
                ? (documentSize / eachCursorSize + 1)
                : documentSize / eachCursorSize;
        log.info("documentCount:{}, connectCursorCount:{}", documentSize, connectCursorTime);

        //循环连接mysql游标，多线程执行遍历数据后导入es
        for (int i = 0; i < connectCursorTime; i++) {
            int cursorCount = i;
            Future<Integer> future = executor.submit(() -> {

                long hasExportCount = cursorCount * eachCursorSize;
                long aimExportCount = (cursorCount + 1) * eachCursorSize > documentSize ? documentSize : (cursorCount + 1) * eachCursorSize;
                int firstPage = (int) (hasExportCount % pageSize > 0 ? hasExportCount / pageSize + 1 : hasExportCount / pageSize);
                int lastPage = (int) (aimExportCount % pageSize > 0 ? aimExportCount / pageSize + 1 : aimExportCount / pageSize);
                List<ElasticEntity> elasticEntityList = new ArrayList<ElasticEntity>();

                for (int page = firstPage; page <= lastPage; page++) {
                    Pageable pageable = PageRequest.of(page, pageSize);
                    Page<Tag> tagPage = tagRepository.findAll(pageable);
                    for (Tag tag : tagPage.getContent()) {
                        TagDTO tagDTO = TagDTO.from(tag);
                        if (tag.getTreeId() != 0) {
                            TagTree tagtree = tagTreeService.getTagTreeById(tag.getTreeId());
                            if (tagtree == null) {
                                log.warn("tagId:{}，标签树不存在，同步es失败", tag.getId());
                                continue;
                            }
                            tagDTO.setTreeName(tagtree.getName());
                        } else {
                            tagDTO.setTreeName("");
                        }

                        Map<String, Object> tagMap = JsonUtils.jsonToObj(JsonUtils.obj2String(tagDTO), new TypeReference<Map<String, Object>>() {
                        });
                        if (tagMap == null) {
                            log.warn("tagId:{}，Json转换有误，同步es失败", tag.getId());
                            continue;
                        }
                        elasticEntityList.add(new ElasticEntity(String.valueOf(tagDTO.getId()), tagMap))
                        ;
                    }
                    // 每页批量异步保存一次
                    esCurdService.insertBatch(TagConstants.TAG_IDX, elasticEntityList);
                    elasticEntityList.clear();

                }
                return cursorCount + 1;
            });
            futures.add(future);
        }

        try {
            log.info("获取mysql数据导入ES 索引 执行结果");
            for (Future<Integer> future : futures) {
                log.info("已执行完的cursor:{}", future.get());
            }
            log.info("mysql数据导入ES 索引执行完毕");
        } catch (Exception e) {
            log.error("执行过程异常：", e);
        }
    }
}
